//文件负责mqtt客户端业务逻辑的交互
package internal

import (
	"encoding/json"
	"github.com/astaxie/beego"
	"sync"
	"t11Lib/t11Errors"
	"t11Lib/t11Log"
	"t11Lib/t11Model"
	"time"
	"亚太-iot-服务器/biz"
	"亚太-iot-服务器/conf"
	"亚太-iot-服务器/models"
)

const GatewayBucketCnt = 100

//网关在内存中维护的状态
type gatewayStatus struct {
	IsOnline bool
}
type gatewayBucket struct {
	dict map[string]*gatewayStatus
	lock *sync.RWMutex
}
type GatewayHub struct {
	buckets  [GatewayBucketCnt]gatewayBucket
	listener *MqttListenerManagerInternal
}

func NewGatewayManager(listener *MqttListenerManagerInternal) *GatewayHub {
	ret := &GatewayHub{buckets: [GatewayBucketCnt]gatewayBucket{}}
	for i := 0; i < GatewayBucketCnt; i++ {
		ret.buckets[i].dict = make(map[string]*gatewayStatus)
		ret.buckets[i].lock = &sync.RWMutex{}
	}
	ret.listener = listener

	if beego.BConfig.RunMode == beego.PROD {
		ret.flushGatewayStatus(time.Minute * 5)
		go ret.flushGatewayStatus(time.Hour * 4)
	} else {
		//测试模式下，只会刷新指定网关的联网状态
		ret.flushGatewayStatus(time.Nanosecond)
		go ret.flushGatewayStatus(time.Nanosecond)
	}

	return ret
}

//更新所有网关的联网状态
//t为总时间
func (self *GatewayHub) flushGatewayStatus(t time.Duration) t11Errors.T11Error {
	gwIds, err := models.GatewayDao.GetAllGatewayIds(t11Model.NewDbConn())
	if err != nil {
		t11Log.Debuglnf("err = %v", err)
		return err
	}

	if beego.BConfig.RunMode == beego.DEV {
		//调试模式下，只刷新给定列表中的设备
		gwIds = conf.AppConfig.TestGatewayIds
	}
	t11Log.Debuglnf("刷新网关状态，gwIds = %v", gwIds)

	if len(gwIds) == 0 {
		return nil
	}
	timeSip := int64(t) / int64(len(gwIds))
	if timeSip <= 0 {
		timeSip = 1
	}
	gwIdMap := make([][]string, MqttThreadCnt)
	for _, id := range gwIds {
		idx := ClientIdToInt(id)
		gwIdMap[idx] = append(gwIdMap[idx], id)
	}

	const batchSize = 10
	for _, ids := range gwIdMap {
		for i := 0; i < len(ids); i += batchSize {
			var gwIds []string
			for j := i; j < len(ids); j++ {
				gwIds = append(gwIds, ids[j])
			}
			//先将联网状态设置成false，然后再发送queryGatewayStatus的命令
			self.SetGatewayListOnlineStatus(false, gwIds...)
			for _, gwId := range gwIds {
				_ = self.SendMsg(&biz.MqttMsg{
					Cmd:       biz.CmdToGateway,
					SubCmd:    "queryGatewayStatus",
					GatewayId: gwId,
				})
				time.Sleep(time.Duration(timeSip))
			}
		}
	}
	return nil
}


func (self *GatewayHub) SetGatewayOnlineStatus(status bool, gatewayId string) {
	idx := ClientIdToInt(gatewayId)
	self.buckets[idx].lock.Lock()
	defer self.buckets[idx].lock.Unlock()
	if val, ok := self.buckets[idx].dict[gatewayId]; ok {
		val.IsOnline = status
	} else {
		self.buckets[idx].dict[gatewayId] = &gatewayStatus{IsOnline: status}
	}
}
func (self *GatewayHub) GetGatewayOnlineStatus(gatewayId string) bool {
	idx := ClientIdToInt(gatewayId)
	self.buckets[idx].lock.RLock()
	defer self.buckets[idx].lock.RUnlock()
	if val, ok := self.buckets[idx].dict[gatewayId]; ok {
		return val.IsOnline
	}
	return false
}

func (self *GatewayHub) SendMsg(obj *biz.MqttMsg) t11Errors.T11Error {
	if obj.GatewayId == "" {
		return t11Errors.New("非法的网关Id")
	}

	data, e := json.Marshal(obj)
	if e != nil {
		return t11Errors.Wrap(e)
	}
	return self.listener.SendData(obj.GatewayId, string(data))
}
//将网关状态设置为离线（可以批量操作，减少上锁）
func (self *GatewayHub) SetGatewayListOnlineStatus(status bool, gwIds ...string) {
	if len(gwIds) == 0 {
		return
	}
	//根据bucket分成不同的list
	gwIdMap := make(map[uint32][]string)
	for _, item := range gwIds {
		idx := ClientIdToInt(item)
		if val, ok := gwIdMap[idx]; ok {
			gwIdMap[idx] = append(val, item)
		} else {
			gwIdMap[idx] = []string{item}
		}
	}
	for k, v := range gwIdMap {
		//每个bucket中的gatewayId一次性设置状态
		func() {
			self.buckets[k].lock.Lock()
			defer self.buckets[k].lock.Unlock()
			dict := self.buckets[k].dict
			for _, id := range v {
				if val, ok := dict[id]; ok {
					val.IsOnline = status
				} else {
					dict[id] = &gatewayStatus{IsOnline: status}
				}
			}
		}()
	}
}
